package com.example.redis.frequency;

import com.alibaba.fastjson.JSONObject;
import com.example.redis.service.JedisLockUtil;
import com.example.redis.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * http 请求的频率 建议每次请求三方，先调用次util 获取当前接口的剩余次数
 */
@Slf4j
@Component
public class HttpFrequencyUtils {

	private static RedisService redisService;

	@Resource
	public void setRedisService(RedisService redisService) {
		HttpFrequencyUtils.redisService = redisService;
	}

	private static final String baseRedisKey = "TOP_HTTP_";

	/**
	 * 从计数开始，到三方接收到接口调用的时间差，是一个不可控的值，且多次调用时，这个值不同。
	 * 这个区别可能造成top和三方在统计调用次数时有偏差，即：有可能造成top认为未超过频率限制，但三方认为已超过频率限制。
	 * 针对这种情况，top将统计调用次数时间窗口额外添加50毫秒以降低这种情况出现的概率
	 */
	private static final int fixedTimeDiff = 50;

	/**
	 * 使用说明：调用此方法 尽可能在 生成请求参数之前调用， 否则可能会造成接口调用失败，例如：提前生成请求参数在等待5分钟后，时间戳校验失败
	 * <p>
	 * 案例1： 饿百新增商品接口 每个商户 每分钟限制10次， 在TOP后台应配置 平台=饿百，
	 * 动态限制=shopId（可自由发挥，多参数组合）,限制时间=60 限制次数=10 url=product.add isSen方法传入
	 * plateformEnum=EB_JK dynamicConstraint=10024(实际的商户id) url=product.add
	 * <p>
	 * 案例2： 百度更新商品接口 10分钟限制20次， 在TOP后台应配置 平台=饿百， 动态限制=(不填即可) 限制时间=60 限制次数=10
	 * url=update.add isSen方法传入 plateformEnum=BD dynamicConstraint=null
	 * url=update.add
	 *
	 * @param url               必填，必须唯一，必须与页面配置的参数一致
	 * @return 是否有次数
	 */
	public static boolean isSend( String url) {
		return isSend( url, null);
	}

	public static boolean isSend(String url, Priority priority) {
		//load db,查询频率配置
		// 查询频率配置
		HttpFrequency hfConfig = new HttpFrequency();
		hfConfig.setRestrictTime(10);
		hfConfig.setRestrictFrequency(10);
		if (hfConfig == null) {
			return true;
		}

		// 获取当前操作的redis key 和接口信息
		String redisKey = getDynamicRedisKey(url, priority);

		Integer period = hfConfig.getRestrictTime();// 限制时间 重试时间
		Integer frequency = hfConfig.getRestrictFrequency();

		// 开始操作
		// 操作前拿锁，是否可以进行操作。
		String lockKey = redisKey + "_lock";
		boolean isLock = false;

		// 重试 {maxLockCount} 次加锁，每次加锁失败sleep 0.1秒，{period} 秒后仍加锁失败的获取次数失败
		int maxLockCount = period * 1000 / 100;
		for (int loopCount = 0; loopCount < maxLockCount; loopCount++) {
			try {
				if (JedisLockUtil.tryLockFriendly(lockKey, 1, period, Thread.currentThread().getId() + "")) {
					// 加锁成功，操作。 锁时间 = 接口限制的时间 N秒
					isLock = true;
					String queueString = (String) redisService.get(redisKey);
					if (priority == null) {
						queueString = awaitForNormalApi(redisKey, queueString, frequency, period);
					} else {
						queueString = awaitForTowPathApi(redisKey, queueString, frequency, period, priority);
					}

					redisService.set(redisKey, queueString);
					return true;
				} else {
					// 加锁失败， 等待后重试，sleep 0.1秒
					try {
						Thread.sleep(100);
					} catch (InterruptedException e) {
					}
				}
			} finally {
				// 避免加锁成功后，操作失败可能会死锁
				if (isLock) {
					JedisLockUtil.unLockFriendly(lockKey, Thread.currentThread().getId() + "");
				}
			}
		}

		return false;
	}

	private static String awaitForTowPathApi(String redisKey, String queueString, Integer frequency, Integer period, Priority priority) {
		//最低频率
		BigDecimal lowTrafficRate = BigDecimal.TEN;
		int lowFrequency = new BigDecimal(frequency).multiply(lowTrafficRate).setScale(0, BigDecimal.ROUND_UP).intValue();
		int highFrequency = frequency - lowFrequency;
		long currentTime = System.currentTimeMillis();
		long elapsedTime = currentTime - period * 1000 - fixedTimeDiff;

		PriorityQueue queue = null;
		if (StringUtils.isEmpty(queueString)) {
			queue = new PriorityQueue(highFrequency, lowFrequency);
		} else {
			queue = JSONObject.parseObject(queueString, PriorityQueue.class);
			discardOldData(queue, Priority.LOW, elapsedTime);
			discardOldData(queue, Priority.HIGH, elapsedTime);

			if (queue.getHighSize() != highFrequency || queue.getLowSize() != lowFrequency) { // 阀值或频率变更，确保每个队列中已有元素不能超过新的阀值
				// resize
				if (lowFrequency < queue.getLowSize()) { // 低优先队列缩小
					int diff = queue.getLowList().size() - lowFrequency;
					if (diff > 0) { // 且低优先队列中已有元素超过了新的阀值
						Long top = null;
						for (int i = 0; i < diff; i++) {
							top = queue.poll(Priority.LOW);
						}
						sleep(redisKey, period, currentTime, top);
						currentTime = System.currentTimeMillis();
					}
				}

				if (highFrequency < queue.getHighSize()) { // 高优先队列缩小
					int diff = queue.getHighList().size() - highFrequency;
					if (diff > 0) { // 且高优先队列中已有元素超过了新的阀值
						Long top = null;
						for (int i = 0; i < diff; i++) {
							top = queue.poll(Priority.HIGH);
						}
						sleep(redisKey, period, currentTime, top);
						currentTime = System.currentTimeMillis();
					}
				}
				queue.setHighSize(highFrequency);
				queue.setLowSize(lowFrequency);
			}
		}

		try {
			queue.add(priority, currentTime);
		} catch (IllegalStateException e) {
			Long top = queue.poll(priority);
			if (top == null) {
				// 添加失败但queue中没有元素，说明这个通道大小为0，所有元素都添加到另一个通道中
				top = queue.poll(priority == Priority.HIGH ? Priority.LOW : Priority.HIGH);
			}
			sleep(redisKey, period, currentTime, top);
			queue.add(priority, System.currentTimeMillis());
		}

		return JSONObject.toJSONString(queue);
	}

	private static void discardOldData(PriorityQueue queue, Priority priority, long elapsedTime) {
		while (true) {
			Long top = queue.peek(priority);
			if (top == null || top > elapsedTime) {
				return;
			} else {
				queue.poll(priority);
			}
		}
	}

	private static void sleep(String redisKey, Integer period, long currentTime, Long top) {
		long elapsedTime = currentTime - period * 1000 - fixedTimeDiff;
		if (top > elapsedTime) {
			long sleepMillis = top - elapsedTime;
			log.info("sleeping {} for {}", sleepMillis, redisKey);
			try {
				Thread.sleep(sleepMillis);
			} catch (InterruptedException e1) {
				log.error(e1.getMessage());
			}
		}
	}

	private static String awaitForNormalApi(String redisKey, String queueString, Integer frequency, Integer period) {
		long elapsedTime = System.currentTimeMillis() - period * 1000 - fixedTimeDiff;

		LinkedBlockingQueue<Long> queue = new LinkedBlockingQueue<>(frequency);
		if (!StringUtils.isEmpty(queueString)) {
			List<Long> timestampList = JSONObject.parseArray(queueString, Long.class);
			for (Long timestamp : timestampList) {
				if (timestamp > elapsedTime) {
					queue.add(timestamp);
				}
			}
		}

		try {
			queue.add(System.currentTimeMillis());
		} catch (IllegalStateException e) {
			Long top = queue.peek();
			if (top > elapsedTime) {
				long sleepMillis = top - elapsedTime;
				log.info("sleeping {} for {}", sleepMillis, redisKey);
				try {
					Thread.sleep(sleepMillis);
				} catch (InterruptedException e1) {
					log.error(e1.getMessage());
				}
			}
			queue.poll();
			queue.add(System.currentTimeMillis());
		}
		return JSONObject.toJSONString(queue);
	}

	//低流量率
	private static BigDecimal getLowTrafficRate() {
		return BigDecimal.TEN;
	}

	private static String getDynamicRedisKey(String url, Priority priority) {
		StringBuilder redisKey = new StringBuilder(baseRedisKey).append("_").append(url);
		if (priority != null) {
			redisKey.append("_p");
		}
		return redisKey.toString();
	}

}